package com.ds.lens.flink.monitor.monitorjob.function;

import com.google.common.collect.Maps;
import com.ds.lens.data.common.constant.MonitorStateEnum;
import com.ds.lens.data.common.document.dto.MonitorEvent;
import com.ds.lens.flink.monitor.common.LogUtils;
import com.ds.lens.flink.monitor.common.MonitorDefinitionFactory;
import com.ds.lens.flink.monitor.common.bo.Metric;
import com.ds.lens.flink.monitor.common.bo.MetricBundle;
import com.ds.lens.flink.monitor.common.bo.MonitorDefinition;
import com.ds.lens.flink.monitor.monitorjob.function.detector.MonitorEventDetector;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.springframework.util.CollectionUtils;

import java.util.Map;

@Slf4j
public class MonitorEventGenerateFunction extends KeyedProcessFunction<Long, MetricBundle, Tuple2<MetricBundle, Map<String, MonitorEvent>>> {


    private static final long serialVersionUID = 1328679547406468464L;

    @Override
    public void processElement(MetricBundle value, Context ctx, Collector<Tuple2<MetricBundle, Map<String, MonitorEvent>>> out) throws Exception {
        MonitorDefinition definition = MonitorDefinitionFactory.get().getMonitorDefinitionByState(value.getState());
        LogUtils.debug("Input param metric bundle={}", value);
        if (definition == null || !definition.getMonitorState().getKey().getVersion().equals(value.getState().getKey().getVersion())) {
            return;
        }
        MonitorEventDetector detector = definition.getDetector();
        Map<String, MonitorEvent> map = Maps.newHashMapWithExpectedSize(value.getMetrics().size());
        for (Metric metric : value.getMetrics().values()) {
            MonitorEvent event = detector.detect(metric);
            if (event.getTriggered().getState() == MonitorStateEnum.NO_DATA) {
                continue;
            }
            map.put(event.getTriggered().getGroupValue(), event);
        }
        if (!CollectionUtils.isEmpty(map)) {
            out.collect(Tuple2.of(value, map));
        }

    }
}
